package belf.migrate.engine.job.jdbc;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.sink.jdbc.helper.ColumnRuleHelper;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.taskconf.*;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.*;
import belf.migrate.api.job.Job;
import com.alibaba.druid.pool.DruidDataSource;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.lang3.StringUtils;

import java.sql.*;
import java.util.*;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class JDBCReader implements Runnable {
    private final Job job;
    private final JobContext jobContext;
    private final TableSyncConf tableSyncConf;
    private final Pipe pipe;
    private final int sliceStart;
    private final int sliceEnd;
    private final ColumnRuleHelper columnRuleHelper;

    private final Task task;
    private final TaskConf taskConf;
    private final SourceConf sourceConf;
    private final Catalog sourceCatalog;
    private final TablePath tablePath;

    private DruidDataSource dataSource;
    private Map<String, ColumnMetaData> columnMap;

    public JDBCReader(Job job, JobContext jobContext, TableSyncConf tableSyncConf, Pipe pipe, int sliceStart, int sliceEnd, ColumnRuleHelper columnRuleHelper) {
        this.job = job;
        this.jobContext = jobContext;
        this.tableSyncConf = tableSyncConf;
        this.pipe = pipe;
        this.sliceStart = sliceStart;
        this.sliceEnd = sliceEnd;
        this.columnRuleHelper = columnRuleHelper;

        this.task = job.getTask();
        this.taskConf = task.getTaskConf();
        this.sourceConf = task.getTaskConf().getSourceConf();
        this.sourceCatalog = jobContext.getSourceCatalog();


        String sourceDatabaseName = sourceConf.getDatabaseName();
        String sourceSchemaName = sourceConf.getSchemaName();
        String tableName = tableSyncConf.getTableName();

        QualifiersModel sourceQualifiersModel = jobContext.getSourceQualifiersModel();

        this.tablePath = TablePath.of(sourceDatabaseName, sourceSchemaName, tableName, sourceQualifiersModel);

        this.dataSource = JDBCJobRunner.initDataSource(sourceConf.getConnectionConf());
        try {
            this.columnMap = sourceCatalog.getTable(this.tablePath).getColumnMap();
            columnRuleHelper.putTableColumnMetaData(this.tablePath.getTableName(), this.columnMap);
        } catch (Exception e) {
            log.error("Exception", e);
        }
    }

    private JdbcExecuteResult readResultSet(PreparedStatement preparedStatement, TablePath tablePath) {
        JdbcExecuteResult jdbcExecuteResult = new JdbcExecuteResult();
        try (ResultSet resultSet = preparedStatement.getResultSet()) {
            List<ColumnMetaData> columns = CatalogUtils.getColumns(resultSet.getMetaData(), tablePath);
            List<Map<String, Object>> rows = new ArrayList<>();
            jdbcExecuteResult.setColumns(columns);
            int columnSize = columns.size();
            while (resultSet.next()) {
                Map<String, Object> resultMap = new LinkedHashMap<>();
                for (int i = 1; i <= columnSize; i++) {
                    String columnName = columns.get(i - 1).getColumnName();
                    Object columnValue = resultSet.getObject(i);
//                            log.info("{} {} = {}", i, columnName, columnValue);
                    resultMap.put(columnName, columnValue);
                }
                rows.add(resultMap);
            }
            jdbcExecuteResult.setRows(rows);
        } catch (SQLException e) {
            log.error("SQLException", e);
        }
        return jdbcExecuteResult;
    }

    /**
     * SELECT column1, column2, ...
     * FROM table_name
     * ORDER BY some_column
     * OFFSET ? ROWS FETCH NEXT ? ROWS ONLY;
     * <p>
     * OFFSET ? ROWS: Skips the specified number of rows before beginning to return rows.
     * FETCH NEXT ? ROWS ONLY: Specifies the number of rows to return after the offset.
     */
    private JdbcExecuteResult findAllByOffsetFetch(TablePath tablePath, int limit, int offset, String sortBy, Object last, boolean desc) throws SQLException {
        JdbcExecuteResult jdbcExecuteResult = null;

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT ");
        sb.append(columnRuleHelper.generate_select_column_list(this.tablePath.getTableName()));
        sb.append(" FROM ");
        sb.append(sourceCatalog.getFullTableName(tablePath));
        String sortByQuoted = sourceCatalog.getQuote() + sortBy + sourceCatalog.getQuote();
        if (StringUtils.isNotBlank(sortBy) && Objects.nonNull(last)) {
            sb.append(" WHERE ");
            sb.append(sortByQuoted);
            sb.append(" > ?");
        }
        if (StringUtils.isNotBlank(sortBy)) {
            sb.append(" ORDER BY ");
            sb.append(sortByQuoted);
            if (desc) {
                sb.append(" DESC");
            }
        }
        sb.append(" OFFSET ? ROWS FETCH NEXT ? ROWS ONLY;");
        String sql = sb.toString();
        log.info("SQL: {}, last: {}, offset: {}, limit: {}", sql, last, offset, limit);
        try (Connection connection = dataSource.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {

            if (StringUtils.isNotBlank(sortBy) && Objects.nonNull(last)) {
                preparedStatement.setObject(1, last);
                preparedStatement.setInt(2, offset);
                preparedStatement.setInt(3, limit);
            } else {
                preparedStatement.setInt(1, offset);
                preparedStatement.setInt(2, limit);
            }

            boolean hasResultSet = preparedStatement.execute();
            if (hasResultSet) {
                return readResultSet(preparedStatement, tablePath);
            }
        }

        return jdbcExecuteResult;
    }


    /**
     * SELECT *
     * FROM (
     * SELECT column1, column2, ...,
     * ROW_NUMBER() OVER (ORDER BY some_column) AS row_num
     * FROM table_name
     * )
     * WHERE row_num BETWEEN start_value AND end_value;
     * <p>
     * ROW_NUMBER() OVER (ORDER BY some_column): Assigns a unique sequential integer to rows, based on the specified column order.
     * start_value and end_value: Define the range of row numbers to retrieve.
     */
    private JdbcExecuteResult findAllByRowNumber(TablePath tablePath, int limit, int offset, String sortBy, Object last, boolean desc) throws SQLException {
        JdbcExecuteResult jdbcExecuteResult = null;
        StringBuffer sb = new StringBuffer();
        sb.append("SELECT ");
        sb.append(columnRuleHelper.generate_select_column_list(this.tablePath.getTableName()));
        sb.append(" FROM (SELECT ");
        sb.append(columnRuleHelper.generate_select_column_list(this.tablePath.getTableName()));
        sb.append(", ROW_NUMBER() OVER (ORDER BY ");
        String sortByQuoted = sourceCatalog.getQuote() + sortBy + sourceCatalog.getQuote();
        sb.append(sortByQuoted);
        sb.append(") AS rn FROM ");
        sb.append(sourceCatalog.getFullTableName(tablePath));
        if (null != last) {
            sb.append(" WHERE ");
            sb.append(sortByQuoted);
            sb.append(" > ?");
        }
        sb.append(") WHERE rn BETWEEN ? AND ?;");
        String sql = sb.toString();
        log.info("SQL: {}, last: {}, BETWEEN {} AND {}", sql, last, offset + 1, offset + limit);
        try (Connection connection = dataSource.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {

            if (null != last) {
                preparedStatement.setObject(1, last);
                preparedStatement.setInt(2, offset + 1);
                preparedStatement.setInt(3, offset + limit);
            } else {
                preparedStatement.setInt(1, offset + 1);
                preparedStatement.setInt(2, offset + limit);
            }

            boolean hasResultSet = preparedStatement.execute();
            if (hasResultSet) {
                return readResultSet(preparedStatement, tablePath);
            }
        }

        return jdbcExecuteResult;
    }

    /**
     * SELECT column1, column2, ...
     * FROM table_name
     * ORDER BY some_column
     * LIMIT count OFFSET start;
     * <p>
     * LIMIT count OFFSET start allows you to specify both the number of rows to retrieve (count) and the number of rows to skip (start).
     */
    private JdbcExecuteResult findAllByLimitOffset(TablePath tablePath, int limit, int offset, String sortBy, Object last, boolean desc) throws SQLException {
        JdbcExecuteResult jdbcExecuteResult = null;

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT ");
        sb.append(columnRuleHelper.generate_select_column_list(this.tablePath.getTableName()));
        sb.append(" FROM ");
        sb.append(sourceCatalog.getFullTableName(tablePath));
        if (StringUtils.isNotBlank(sortBy)) {
            String sortByQuoted = sourceCatalog.getQuote() + sortBy + sourceCatalog.getQuote();
            sb.append(" WHERE ");
            sb.append(sortByQuoted);
            sb.append(" > ?");
            sb.append(" ORDER BY ");
            sb.append(sortByQuoted);
            if (desc) {
                sb.append(" DESC");
            }
        }
        sb.append(" LIMIT ? OFFSET ?;");
        String sql = sb.toString();
        log.info("SQL: {}, last: {}, limit: {}, offset: {}", sql, last, limit, offset);
        try (Connection connection = dataSource.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {

            if (StringUtils.isNotBlank(sortBy)) {
                preparedStatement.setObject(1, last);
                preparedStatement.setInt(2, limit);
                preparedStatement.setInt(3, offset);
            } else {
                preparedStatement.setInt(1, limit);
                preparedStatement.setInt(2, offset);
            }

            boolean hasResultSet = preparedStatement.execute();
            if (hasResultSet) {
                return readResultSet(preparedStatement, tablePath);
            }
        }

        return jdbcExecuteResult;
    }

    @Override
    public void run() {
        try {
            int pageNumber = 1;
            int confPageSize = EngineConf.getInstance().getInt("jdbc.reader.page.size", 100);
            int pageSize = tableSyncConf.getPageSize() != null ? tableSyncConf.getPageSize() : confPageSize;

            if (tableSyncConf.getSyncType() == TableSyncConf.SyncType.INCREMENT_SYNC) {  // 增量同步必须设置sortBy及last
                if (null == tableSyncConf.getSortBy() || null == tableSyncConf.getLast()) {
                    String error_msg = String.format("Table:%s INCREMENT_SYNC(%s) MUST Set sortBy(%s) And last(%s)", tablePath.getTableName(), tableSyncConf.getSyncType(), tableSyncConf.getSortBy(), tableSyncConf.getLast());
                    PostLog.sendLogModel(job.getJobId(), job.getTask().getTaskType().getName(), error_msg, LogLevelType.ERROR.getType(), FAILTURE,
                            jobContext.getProgress());
                    log.error(error_msg);
                    return;
                }
            }

            while (true) {
                int offset = sliceStart + (pageNumber - 1) * pageSize;
                if (offset >= sliceEnd) {  // 偏移已超出分片结束位置
                    log.info("offset >= sliceEnd, JDBCReader {} break, sliceStart: {}, sliceEnd: {}, offset: {} Finish.", Thread.currentThread().getName(), sliceStart, sliceEnd, offset);
                    break;
                }
                int limit = Math.min(pageSize, sliceEnd - offset);  // 最后一页不足量

                JdbcExecuteResult result;
                if (tableSyncConf.getSyncType() == TableSyncConf.SyncType.FULL_SYNC) {
                    if (sourceConf.getCatalogName().equals(DatabaseIdentifier.ORACLE)) {
                        result = findAllByRowNumber(tablePath, limit, offset, tableSyncConf.getSortBy(), null, false);
                    } else if (sourceConf.getCatalogName().equals(DatabaseIdentifier.MYSQL)) {
                        result = findAllByLimitOffset(tablePath, limit, offset, null, null, false);
                    } else {
                        result = findAllByOffsetFetch(tablePath, limit, offset, tableSyncConf.getSortBy(), null, false);
                    }
                    log.info("Table: {}, sliceStart: {}, sliceEnd: {}, limit: {}, offset: {}, Got rows: {}", tablePath.getTableName(), sliceStart, sliceEnd, limit, offset, result.getRows().size());
                } else {
                    if (sourceConf.getCatalogName().equals(DatabaseIdentifier.ORACLE)) {
                        result = findAllByRowNumber(tablePath, limit, offset, tableSyncConf.getSortBy(), tableSyncConf.getLast(), false);
                    } else if (sourceConf.getCatalogName().equals(DatabaseIdentifier.MYSQL)) {
                        result = findAllByLimitOffset(tablePath, limit, offset, tableSyncConf.getSortBy(), tableSyncConf.getLast(), false);
                    } else {
                        result = findAllByOffsetFetch(tablePath, limit, offset, tableSyncConf.getSortBy(), tableSyncConf.getLast(), false);
                    }
                    log.info("Table: {}, sliceStart: {}, sliceEnd: {}, limit: {}, offset: {}, sortBy: {}, last: {} Got rows: {}", tablePath.getTableName(), sliceStart, sliceEnd, limit, offset, tableSyncConf.getSortBy(), tableSyncConf.getLast(), result.getRows().size());
                }

                if (result.getRows().isEmpty()) {  // 查询结果为空
                    log.info("result.getRows().isEmpty(), JDBCReader {} break", Thread.currentThread().getName());
                    break;
                }

                for (ColumnMetaData columnMetaData : result.getColumns()) {
//                    log.debug("Column {}", columnMetaData);
                }


                for (Map<String, Object> row : result.getRows()) {
                    for (Map.Entry<String, Object> entry : row.entrySet()) {
                        Object rv = columnRuleHelper.apply_read_rule(tablePath.getTableName(), entry.getKey(), entry.getValue());
//                        log.debug("Table: {}, Column: {}, Original: {}, New: {}", tablePath.getTableName(), entry.getKey(), entry.getValue(), rv);
                        entry.setValue(rv);
                    }
                }

                pipe.put(result);
                pipe.readCount(result.getRows().size());
                if (pipe.getRead() == pipe.getTotal()) {
                    log.info(pipe.getStatus());
                    PostLog.sendLogModel(job.getJobId(), job.getTask().getTaskType().getName(), pipe.getStatus(), LogLevelType.INFO.getType(), SUCCESS,
                            jobContext.getProgress());
//                  TODO 结束读线程组 readThreadPool.shutdown();
                    log.info("pipe.getRead() == pipe.getTotal(), JDBCReader {} break", Thread.currentThread().getName());
                    break;
                }
                pageNumber += 1;
            }
        } catch (Exception e) {
            log.error("Exception", e);
            PostLog.sendLogModel(job.getJobId(), job.getTask().getTaskType().getName(), String.format("JDBC Reader Exception: %s", e.getMessage()), LogLevelType.FATAL.getType(), FAILTURE,
                    jobContext.getProgress());
        }
    }

    /*
    全量，行数，大字段
    增量，位置记录
    数据量大小，超大表OFFSET扫描并跳过性能非常低效，使用主键或唯一索引代替OFFSET

        MySQL/PostgreSQL: LIMIT ? OFFSET ?，PostgreSQL和MySQL 8.0 版本开始也支持ROW_NUMBER()函数
        Oracle: 使用 ROWNUM，自Oracle 7几乎所有的Oracle数据库版本，ROWNUM是在结果集返回行的过程中分配的伪列，需要先排序后取行，即使用嵌套查询：SELECT * FROM (SELECT * FROM Employees ORDER BY EmployeeID) WHERE ROWNUM <= 10;
                或 ROW_NUMBER() OVER (ORDER BY column)，ROW_NUMBER() 函数在 Oracle 9i 及更高版本中受支持，但在更早版本中并不支持，使用 ROW_NUMBER() 时，ORDER BY 是必需的，以确保你得到的行号是按照你期望的顺序分配的
                从 Oracle 12c 版本开始，你还可以使用更简单的分页方法，比如 OFFSET ... FETCH，这些方法在功能上与 ROW_NUMBER() 类似，但语法更简洁
        SQL Server: 使用 OFFSET 和 FETCH NEXT，从 SQL Server 2012 开始支持；也可以利用ROW_NUMBER()函数实现分页

        connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
        preparedStatement.setFetchSize(pageSize);
    本地，MemQueue，FileQueue
    远程，Redis, Kafka
    分区，SELECT * FROM sales PARTITION (p2022);
     */

    public void close() {
        dataSource.close();
    }
}
